package com.cennavi.funcs;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Properties;

import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.api.java.function.VoidFunction2;
import org.apache.spark.streaming.Time;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;
import scala.Tuple2;

/** * @author  作者 :allen
    * @date 创建时间：2017年9月20日 上午9:00:39 
    * @version 1.0 
    */

public class RticFlow2kafkaFunction2 implements VoidFunction2<JavaPairRDD<String, String>, Time> {

	private static final long serialVersionUID = 1L;

	private String zookeeper_connect = null;
	private String metadata_broker_list = null;
	private String outTopic = null;
	
	public void call(JavaPairRDD<String, String> arg0, Time arg1) throws Exception {
		if(null != arg0){
			final long messageSize = arg0.count();
			final String timeNow = getHourMin(arg1);
			//控制整5分钟输出
			if(Integer.parseInt(timeNow.substring(14, 16)) % 5 == 0){
				//System.out.println("messageSize = " + messageSize);
				arg0.foreachPartition(new VoidFunction<Iterator<Tuple2<String, String>>>() {

					private static final long serialVersionUID = 1L;

					public void call(Iterator<Tuple2<String, String>> arg0) throws Exception {
						Producer<String, String> producer = createProducer();
						List<KeyedMessage<String, String>> datas = new ArrayList<KeyedMessage<String, String>>();
						int count = 0;
						StringBuilder sb = new StringBuilder();
						System.out.println("currentThread is " + Thread.currentThread().getName());
						while (arg0.hasNext()) {
							count ++;
							String message = arg0.next()._2;
							String prefix = message.substring(0, message.indexOf(","));
							String suffix = message.substring(message.indexOf(",") + 1, message.length());
							suffix = suffix.substring(suffix.indexOf(",") + 1, suffix.length());
							message = prefix + "," + timeNow + "," + suffix;
							String time = message.split(",")[1].replaceAll("-", "").replace(":", "").replace(" ", "");
							//String capacity = message.split(",")[10];
							
							//if(!capacity.equals("null") && !capacity.substring(0, 1).equals("R")){
							sb.append(message);
							sb.append("|");
							//每逢满2000条或者到达最后一条消息时添加到list里
							if(count % 2000 == 0 || !arg0.hasNext()){
								KeyedMessage<String, String> data = new KeyedMessage<String, String>(outTopic, ",,," + time, sb.toString());
								System.out.println("linesNum = " + sb.toString().split("\\|").length);
								datas.add(data);
								sb.setLength(0);
							}
								
							//}
							
						}
						//System.out.println("count = " + count);
						//System.out.println("datas.size = " + datas.size());
						producer.send(datas);
					}});
			}
		}
	}
	
	//构造函数
	public RticFlow2kafkaFunction2(String zookeeper_connect, String metadata_broker_list, String outTopic){
		this.zookeeper_connect = zookeeper_connect;
		this.metadata_broker_list = metadata_broker_list;
		this.outTopic = outTopic;
	}
	
	public Producer<String, String> createProducer(){
		Properties properties = new Properties();  
        properties.put("zookeeper.connect", this.zookeeper_connect);//声明zk  
        properties.put("serializer.class", StringEncoder.class.getName());  
        properties.put("metadata.broker.list", this.metadata_broker_list);// 声明kafka broker  
        return new Producer<String, String>(new ProducerConfig(properties));  
	}

	private String getHourMin(Time time) {
		java.util.Date d = new Date();
		d.setTime(time.milliseconds());
		String dateTime = new SimpleDateFormat("yyyy-MM-dd HH:mm").format(d);
		return dateTime;
	}
	public static void main(String[] args) throws ParseException {
//		String date1 = "2017-09-25 7:00";
//		String date2 = "2020-10-08 7:00";
//		long deltaDays = (new SimpleDateFormat("yyyy-MM-dd HH:mm").parse(date2).getTime()
//				- new SimpleDateFormat("yyyy-MM-dd HH:mm").parse(date1).getTime()) / (1000 * 60 * 60 * 24);
//		System.out.println(deltaDays);
		
//		List<String> test = new ArrayList<String>();
//		test.add("aaa");
//		test.add("bbb");
//		test.add("ccc");
//		Iterator it = test.iterator();
//		int count = 0;
//		while(it.hasNext()){
//			count ++;
//			it.next();
//			if(!it.hasNext()){
//				System.out.println(count);
//			}
//		}
//		StringBuilder sb = new StringBuilder();
//		sb.append("first");
//		sb.setLength(0);
//		System.out.println(sb.toString());
//		sb.append("second");
//		System.out.println(sb.toString());

//		System.out.println(0 % 5);
//		System.out.println(Integer.parseInt("00"));
//		System.out.println(Integer.parseInt("05"));
		String tempStr = "012,45,789";
		String prefix = tempStr.substring(0, tempStr.indexOf(","));
		String suffix = tempStr.substring(tempStr.indexOf(",") + 1, tempStr.length());
		suffix = suffix.substring(suffix.indexOf(",") + 1, suffix.length());
		System.out.println(prefix);
		System.out.println(suffix);
	}
}
